// Package checksum worker_source.go 定义源数据库校验工作线程
package checksum

import (
	"context"      // 导入 context 包，用于控制goroutine生命周期和超时
	"database/sql" // 导入 database/sql 包，用于数据库操作
	"errors"       // 导入 errors 包，用于创建和处理错误
	"fmt"          // 导入 fmt 包，用于格式化字符串
	"sync/atomic"  // 导入 atomic 包，用于原子操作（如计数器）
	"time"         // 导入 time 包，用于处理时间

	log "github.com/sirupsen/logrus" // 导入 logrus 日志库

	"px-table-checksum-go/config" // 导入配置包
	"px-table-checksum-go/db"     // 导入数据库操作包
	"px-table-checksum-go/queue"  // 导入队列接口包
	"px-table-checksum-go/util"   // 导入 util 包以使用 KeyString 相关的工具函数
)

// RunSourceChecksumWorker 运行源数据库校验工作线程
//
// 该函数负责处理单个表的校验任务。它按数据块 (chunk) 从源数据库读取数据，
// 计算每个块的CRC32校验和，将结果存入校验数据库，并将任务信息（用于目标端）放入队列。
// 当一个表的所有块处理完毕或遇到不可恢复的错误时，它会向队列发送一个结束信号，
// 并（如果成功）原子地增加已完成表的计数器。
//
// 参数:
//   - sourceDB: 源数据库连接池 (*sql.DB)
//   - checksumDB: 校验数据库连接池 (*sql.DB)
//   - taskQueue: 任务队列 (实现了 queue.Queue 接口)，用于与目标端 worker 通信
//   - finishedTableCounter: 已完成表的原子计数器 (用于追踪整体进度)
//   - schemaName: 要处理的数据库模式名
//   - tableName: 要处理的表名
//   - ctx: 父上下文 (context.Context)，用于接收取消信号
//
// 返回值:
//   - error: 如果处理过程中发生不可恢复的错误，则返回错误；否则返回 nil
func RunSourceChecksumWorker(sourceDB *sql.DB, checksumDB *sql.DB, taskQueue queue.Queue, finishedTableCounter *atomic.Int32, schemaName, tableName string, ctx context.Context) error {
	// 创建带有上下文信息的日志记录器，方便追踪特定表和 worker 的日志
	logger := log.WithFields(log.Fields{
		"schema": schemaName,
		"table":  tableName,
		"type":   "source", // 标记为源端 worker
	})

	// 标记是否成功完成，用于 defer 中判断是否增加计数器
	var completedSuccessfully bool = false
	// 使用 defer 确保在函数退出时执行清理逻辑
	defer func(start time.Time) {
		// 如果处理成功完成，原子地增加已完成表的计数器
		if completedSuccessfully {
			finishedTableCounter.Add(1) // 原子操作，线程安全
			logger.Info("表处理完成并增加完成计数")
		} else {
			// 如果未成功完成（例如出错或被取消），记录警告
			logger.Warn("表处理未成功完成，不增加完成计数")
		}
		// 记录处理该表的总耗时
		logger.Infof("表处理结束，总耗时: %v", time.Since(start))
	}(time.Now()) // 记录开始时间

	logger.Info("启动源端校验")

	// 获取源库的标识符，用于写入校验结果
	sourceDBID := config.GlobalConfig.Options.SourceDBID
	if sourceDBID == "" {
		logger.Warn("源数据库 ID (SourceDBID) 未在配置中设置，将使用 Source DB 的 host:port")
		sourceDBID = config.GlobalConfig.DBSource.ID()
	}
	logger.Debugf("源数据库 ID: %s", sourceDBID)

	// 1. 构建分块查询的 SQL 模板和获取表的键信息
	sqlTemplate, numKeyColumns, keyInfo, err := db.BuildChunkSQL(sourceDB, schemaName, tableName)
	if err != nil {
		logger.Errorf("构建查询SQL失败: %v", err)
		_ = sendFinishSignal(taskQueue, schemaName, tableName, fmt.Sprintf("构建查询SQL失败: %v", err))
		return fmt.Errorf("构建查询SQL失败: %w", err)
	}
	if keyInfo == nil || len(keyInfo.Columns) == 0 {
		// GetBestUniqueKey 理论上应该返回有效键或错误，这里是双重检查。
		errMsg := "未能获取到表的有效键信息，无法进行基于键的校验"
		logger.Error(errMsg)
		_ = sendFinishSignal(taskQueue, schemaName, tableName, errMsg)
		return errors.New(errMsg)
	}

	logger.Debugf("使用查询模板: %s", sqlTemplate)
	logger.Debugf("键信息: %+v", keyInfo)

	// 2. 初始化分块处理的状态变量
	var currentStartKey string
	var chunkNo int = 1
	var taskErrors []error
	var allTaskCount int
	var errorTaskCount int

	// 获取第一个块的起始键 ("0" 或 "0,0" 等)
	// Modify: Use util.GetInitialStartKey
	currentStartKey = util.GetInitialStartKey(numKeyColumns)
	logger.Debugf("初始起始键: %s", currentStartKey)

	// 3. 循环处理数据块
	for {
		// 在每次循环开始时检查父上下文是否已取消
		select {
		case <-ctx.Done():
			logger.Warnf("源端处理表 %s.%s 在处理 chunk %d 前被取消: %v", schemaName, tableName, chunkNo, ctx.Err())
			_ = sendFinishSignal(taskQueue, schemaName, tableName, fmt.Sprintf("处理被取消: %v", ctx.Err()))
			return ctx.Err()
		default:
			// 继续执行
		}

		// 解析当前块的起始键字符串为 interface{} 切片
		// Modify: Use util.ParseKeyFromString
		startKeyValues, parseErr := util.ParseKeyFromString(currentStartKey, numKeyColumns)
		if parseErr != nil {
			errMsg := fmt.Sprintf("解析起始键值 '%s' 失败: %v", currentStartKey, parseErr)
			logger.Error(errMsg)
			taskErrors = append(taskErrors, errors.New(errMsg))
			_ = sendFinishSignal(taskQueue, schemaName, tableName, errMsg)
			break
		}

		task := &db.ChecksumTask{
			DBID:           sourceDBID,
			SchemaName:     schemaName,
			TableName:      tableName,
			ChunkNo:        chunkNo,
			StartKeyString: currentStartKey,
			StartKeyValues: startKeyValues, // 传递解析后的键值切片
			SQLTemplate:    sqlTemplate,
			NumKeyColumns:  numKeyColumns,
			KeyInfo:        keyInfo,
		}

		allTaskCount++
		chunkLogger := logger.WithField("chunk", chunkNo)
		chunkLogger.Debugf("开始处理块: 起始键=%s", currentStartKey)

		// 4. 执行实际的校验和计算
		calcStartTime := time.Now()
		result, isLastChunk, calcErr := CalculateChunkChecksum(sourceDB, task)
		calcTime := time.Since(calcStartTime)

		// 检查计算是否出错
		if calcErr != nil {
			errorTaskCount++
			errMsg := fmt.Sprintf("计算块 %d 校验和失败: %v", chunkNo, calcErr)
			chunkLogger.Error(errMsg)
			taskErrors = append(taskErrors, errors.New(errMsg))

			// 检查是否达到最大错误数限制
			maxErrors := config.GlobalConfig.Options.MaxErrors
			if maxErrors > 0 && errorTaskCount > maxErrors {
				logger.Errorf("错误数(%d)超过最大允许值(%d)，放弃表 %s.%s 的剩余处理",
					errorTaskCount, maxErrors, schemaName, tableName)
				_ = sendFinishSignal(taskQueue, schemaName, tableName, "错误数超过限制")
				break // 达到错误上限，退出循环
			}

			// 尝试从部分结果中恢复 MaxKey 以处理下一个块 (仅在获取到有效 MaxKey 且不是当前 StartKey 时尝试)
			if result != nil && result.MaxKey != "" && result.MaxKey != currentStartKey {
				chunkLogger.Warnf("计算块 %d 失败，但获取到 MaxKey '%s'，尝试继续下一个块", chunkNo, result.MaxKey)
				currentStartKey = result.MaxKey // 更新起始键
				chunkNo++
				continue // 继续下一个循环
			} else {
				// 如果计算失败且无法获取有效的下一个块的起始键，则无法继续
				chunkLogger.Error("计算校验和失败且无法获取下一块的起始键，表处理中止")
				_ = sendFinishSignal(taskQueue, schemaName, tableName, "计算失败且无法恢复MaxKey")
				break // 无法恢复，退出循环
			}
		}

		// 校验计算成功 (result 不为 nil)
		// 5. 将校验结果写入校验数据库
		writeStartTime := time.Now()
		if dbErr := db.InsertChecksumResult(checksumDB, result); dbErr != nil {
			errorTaskCount++
			errMsg := fmt.Sprintf("保存块 %d 校验结果失败: %v", chunkNo, dbErr)
			chunkLogger.Error(errMsg)
			taskErrors = append(taskErrors, errors.New(errMsg))

			// 检查是否达到最大错误数限制
			maxErrors := config.GlobalConfig.Options.MaxErrors
			if maxErrors > 0 && errorTaskCount > maxErrors {
				logger.Errorf("错误数(%d)超过最大允许值(%d)，放弃表 %s.%s 的剩余处理",
					errorTaskCount, maxErrors, schemaName, tableName)
				_ = sendFinishSignal(taskQueue, schemaName, tableName, "错误数超过限制")
				break // 达到错误上限，退出循环
			}
			// 保存结果失败，但仍然可以尝试处理下一个块 (如果 MaxKey 有效)
		}
		writeTime := time.Since(writeStartTime)

		// 6. 将任务信息放入队列，供目标端 worker 处理
		if queueErr := taskQueue.Put(task); queueErr != nil {
			chunkLogger.Errorf("发送任务到队列失败: %v", queueErr)
			if errors.Is(queueErr, queue.ErrQueueClosed) {
				logger.Warn("尝试发送任务时发现队列已关闭")
				return queue.ErrQueueClosed // 队列关闭，worker 应该退出
			}
			_ = sendFinishSignal(taskQueue, schemaName, tableName, fmt.Sprintf("发送任务到队列失败: %v", queueErr))
			return fmt.Errorf("发送任务到队列失败: %w", queueErr)
		}

		// 记录当前块处理成功的日志
		chunkLogger.Infof("完成: 行数=%d, CRC32=%d, 计算=%v, 写入=%v",
			result.RowCount, result.CRC32, calcTime, writeTime)

		// 7. 检查是否是最后一个块 (重要修复点)
		// 如果 CalculateChunkChecksum 标记为最后一个块，则应该退出循环。
		// 即使返回了行数 < ChunkSize，只要 isLastChunk 为 true，就表示这是最后一个块。
		// 对于空表，result.RowCount == 0 且 isLastChunk == true，应该在此处退出。
		if isLastChunk {
			logger.Info("检测到表的最后一个块或无法继续，表处理完成")
			break // 退出循环
		}

		// 准备下一个块的起始键 (仅当不是最后一个块时更新)
		// 必须确保 MaxKey 有效且与当前 StartKey 不同，否则可能死循环。
		if result.MaxKey == "" || result.MaxKey == currentStartKey {
			// 这通常不应该发生，除非是空块 (已在 isLastChunk = true时处理) 或查询逻辑有问题
			chunkLogger.Errorf("计算成功但返回的 MaxKey 为空或与 StartKey 相同 ('%s'), 表处理中止。", currentStartKey)
			_ = sendFinishSignal(taskQueue, schemaName, tableName, fmt.Sprintf("计算成功但MaxKey无效: %s", currentStartKey))
			break // 无法确定下一块，退出循环
		}
		currentStartKey = result.MaxKey // 更新起始键

		// 增加块编号，准备处理下一个块
		chunkNo++
	} // 结束 for 循环

	// 8. 处理循环结束后的状态
	var errorSummary string
	if len(taskErrors) > 0 {
		errorSummary = fmt.Sprintf("%d项任务中有%d项失败，最后错误: %v",
			allTaskCount, errorTaskCount, taskErrors[len(taskErrors)-1])
		logger.Warn(errorSummary)
		if err := sendFinishSignal(taskQueue, schemaName, tableName, errorSummary); err != nil {
			logger.Errorf("发送带错误的结束信号失败: %v", err)
			if errors.Is(err, queue.ErrQueueClosed) {
				return queue.ErrQueueClosed
			}
			return fmt.Errorf("发送结束信号失败: %w (原始错误: %v)", err, taskErrors[0])
		}
		// 返回第一个遇到的错误作为代表
		return fmt.Errorf("处理表时发生错误: %w (共%d/%d项失败)",
			taskErrors[0], errorTaskCount, allTaskCount)
	} else if ctx.Err() != nil {
		logger.Warnf("表 %s.%s 处理被取消: %v", schemaName, tableName, ctx.Err())
		_ = sendFinishSignal(taskQueue, schemaName, tableName, fmt.Sprintf("处理被取消: %v", ctx.Err()))
		return ctx.Err()
	} else {
		logger.Info("所有块处理完毕")
		completedSuccessfully = true // 标记为成功完成
		logger.Debugf("发送结束信号 (无错误)")
		if err := sendFinishSignal(taskQueue, schemaName, tableName, ""); err != nil {
			logger.Errorf("发送成功的结束信号失败: %v", err)
			if errors.Is(err, queue.ErrQueueClosed) {
				return queue.ErrQueueClosed
			}
			return fmt.Errorf("发送结束信号失败: %w", err)
		}
	}

	return nil
}

// RunSourceTablesSplitWorker (已废弃 - 功能合并到 manager.processConfiguredTablesWithContext)
// Deprecated: This function's logic is now part of checksum.processConfiguredTablesWithContext.
func RunSourceTablesSplitWorker(taskQueue queue.Queue, finishedTableCounter *atomic.Int32) (int, error) {
	log.Warn("RunSourceTablesSplitWorker is deprecated and should not be called directly.")
	// 返回 0 和一个错误，表示此函数不应再使用
	return 0, errors.New("RunSourceTablesSplitWorker is deprecated")
}

// sendFinishSignal 发送处理完成信号到目标端队列
// 这是一个内部辅助函数，用于通知目标端某个表的处理已结束（无论成功还是失败）。
func sendFinishSignal(taskQueue queue.Queue, schemaName, tableName, errorMsg string) error {
	// 创建一个特殊的 ChecksumTask 作为结束信号
	// ChunkNo = -1 是约定的结束标志
	finishTask := &db.ChecksumTask{
		SchemaName: schemaName,
		TableName:  tableName,
		ChunkNo:    -1,       // 特殊值表示结束
		Error:      errorMsg, // 携带错误信息（如果处理失败）
	}

	// 记录将要发送的信号信息
	logger := log.WithFields(log.Fields{
		"schema": schemaName,
		"table":  tableName,
	})
	if errorMsg != "" {
		logger.Debugf("向队列发送表 %s.%s 的【错误】结束信号: %s", schemaName, tableName, errorMsg)
	} else {
		logger.Debugf("向队列发送表 %s.%s 的【正常】结束信号", schemaName, tableName)
	}

	// 将结束信号放入队列
	err := taskQueue.Put(finishTask)
	if err != nil {
		// 如果放入队列失败，记录错误
		logger.Errorf("发送结束信号到队列失败 (表: %s.%s): %v", schemaName, tableName, err)
	} else {
		// 记录成功发送
		logger.Debugf("成功发送表 %s.%s 的结束信号", schemaName, tableName)
	}
	// 返回放入队列操作的错误状态
	return err
}
